package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Demo5EventTime {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        /*
         * java,1705627456000
         * java,1705627457000
         * java,1705627458000
         * java,1705627459000
         * java,1705627460000
         * java,1705627465000
         * java,1705627466000
         * java,1705627467000
         * java,1705627470000
         * java,1705627468000
         * java,1705627475000
         */

        //读取数据
        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //解析数据
        DataStream<Tuple2<String, Long>> wordAndTSDS = linesDS.map(line -> {
            String[] split = line.split(",");
            String word = split[0];
            long ts = Long.parseLong(split[1]);
            return Tuple2.of(word, ts);
        }, Types.TUPLE(Types.STRING, Types.LONG));

        //设置事件时间字段和水位线生产策略
        DataStream<Tuple2<String, Long>> assignDS = wordAndTSDS
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                //1、默认水位线等于最新一条数据的时间戳，如果数据乱序会丢失数据
                                //.<Tuple2<String, Long>>forMonotonousTimestamps()
                                //2、水位线前移5秒。延迟5秒计算
                                .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                //指定时间字段
                                .withTimestampAssigner((kv, ts) -> kv.f1)
                );


        /*
         * 每隔5秒统计最近5秒单词的数量  -- 滚动窗口
         * SlidingProcessingTimeWindows： 滑动的处理时间窗口
         * SlidingEventTimeWindows： 滑动的事件时间窗口
         * TumblingProcessingTimeWindows： 滚动的处理时间窗口
         * TumblingEventTimeWindows: 滚动的事件时间窗口
         */

        assignDS
                //转换成kv
                .map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT))
                //按照单词分组
                .keyBy(kv -> kv.f0)
                //每隔5秒计算一次
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                //统计单词的数量
                .sum(1)
                .print();


        env.execute();
    }
}
